package cn._51doit.flink.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * 使用OperatorState需要实现CheckpointedFunction接口,必须实现两个方法initializeState和snapshotState
 *
 * 方法的调用顺序（方法的生命周期）
 *
 * initializeState(一次) -> open(一次)  -> run(一次)  ---> snapshotState(每checkpoint一次会调用一次)  -> cancel(一次) ->  close(一次)
 *
 */
public class MyAtLeastOnceSource extends RichParallelSourceFunction<String> implements CheckpointedFunction {

    private boolean flag = true;

    //在Flink中建议State用transient关键字修饰
    //状态不能认为在序列化时传值，而是必须通过open方法或initializeState方法恢复数据
    private transient ListState<Long> listState;

    private Long offset = 0L;

    private String path;

    public MyAtLeastOnceSource(String path) {
        System.out.println("Constructor Method Invoked！！！");
        this.path = path;
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("initializeState Method Invoked@@@@");
        //该方法是用来初始化或恢复OperatorState
        ListStateDescriptor<Long> stateDescriptor = new ListStateDescriptor<>("offset-state", Long.class);
        listState = context.getOperatorStateStore().getListState(stateDescriptor);
        if(context.isRestored()) {
            offset = listState.get().iterator().next();
        }
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {

        //已经约定好了，一个subtask读取一个文件(例如0号subtask读0.txt)
        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
        RandomAccessFile file = new RandomAccessFile(path + "/" + indexOfThisSubtask + ".txt", "r");
        file.seek(offset);
        while (flag) {
            String line = file.readLine();
            if (line != null) {
                line = new String(line.getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
                synchronized (ctx.getCheckpointLock()) {
                    ctx.collect(indexOfThisSubtask + " --> " + line);
                    offset = file.getFilePointer();
                }
            } else {
                Thread.sleep(1000);
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        listState.clear();
        listState.add(offset);
    }

    @Override
    public void cancel() {
        flag = false;
    }
}
